package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.ServerInfo;
import com.aotain.project.apollo.SiteEvaluateMain;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * 基于UDII数据的流量统计（包大小统计KB)
 * @author Administrator
 *
 */
public class UDIIBaseStatSumBolt extends AbstractBolt{

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	/**
	 * 流量来源统计
	 */
	HashMap<String,HashMap<String,Long>> _mapSourceArea 
		= new HashMap<String,HashMap<String,Long>>();

	HashMap<String,ServerInfo> _mapServers = null;
	HashMap<String,Long> DestIPOCTETS = new HashMap<String,Long>();
	HashMap<String,Long> DestIPPACKET = new HashMap<String,Long>();
	
	long TotalPV = 0L;
	long _updateTime = 0L;
	//HashSet<String> _ImportIPSet = new HashSet<String>();

	
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("IDC","TYPE","PV", "PACKET"));
		declarer.declareStream("DELAY", new Fields("IDC","IP","TYPE","OCTETS", "PACKET","PV"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		//
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}
	
	
	private void countInLocal(Tuple tuple) {
		
		String gisstr = "";
		try
		{
		
			String DestIP = tuple.getString(0);
	//		int destport = Integer.parseInt(tuple.getString(1));
			
			String idc = tuple.getStringByField("idcno");
			
	//		String sourceip = tuple.getString(2);
	//		String sport = tuple.getString(3);
	//		System.err.println("gis64========" + tuple.getString(4));
			byte[] gisbytes =  Base64.decodeBase64(tuple.getString(4));
			gisstr = new String(gisbytes);
			//destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry;
			//destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName;
			String[] gisArray = gisstr.split("#",-1);
	//		String destAreaName = gisArray[0]; //如果地市为空，精确到国家
	//		String DestGeo = gisArray[1];
	//		String sourceAreaName = gisArray[2];  //如果地市为空，精确到国家
	//		String SourceGeo = gisArray[3];
	//		String sourceAreaCountry = gisArray[4];
			String sourceProvinceID =  gisArray[5];
			String sourceProvinceName =  gisArray[6]; //如果省为空，精确到国家
			
			//上行流量字节数
			
			long nUpStreamOctets = tuple.getLongByField("upstreamoctets");
			long nUpStreamPacket = tuple.getLongByField("upstreampacket");
	
			//String dip = tuple.getString(0);
			
			String key = String.format("%s|%s", idc, DestIP);
	
			if(DestIPOCTETS.containsKey(key)) {
				long dipPV = 0L;
				dipPV = DestIPOCTETS.get(key) + (long)nUpStreamOctets;
				DestIPOCTETS.put(key, dipPV);
				
				long dipPacket = 0L;
				dipPacket = DestIPPACKET.get(key) + nUpStreamPacket;
				DestIPPACKET.put(key, dipPacket);
			} else {
				DestIPOCTETS.put(key, (long)nUpStreamOctets);
				DestIPPACKET.put(key, (long)nUpStreamPacket); 
			}
			
			TotalPV++;
			
			//区域分布统计
			String sourceKey = sourceProvinceID + "@" + sourceProvinceName;
			if(_mapSourceArea.containsKey(key)) {
				
				HashMap<String,Long> sourceArea = _mapSourceArea.get(key);
				
				if(sourceArea.containsKey(sourceKey)) {
					long sourcePV = 0L; 
					sourcePV = sourceArea.get(sourceKey) + (long)nUpStreamOctets;
					sourceArea.put(sourceKey, sourcePV);
				} else {
					sourceArea.put(sourceKey, (long)nUpStreamOctets);
				}	
			} else {
				HashMap<String,Long> sourceArea = new HashMap<String,Long>();
				sourceArea.put(sourceKey, (long)nUpStreamOctets);
				_mapSourceArea.put(key, sourceArea);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			Logger.getRootLogger().error(e);
			throw new FailedException("UDIIBaseStatSumBolt-countInLocal Exception,GIS STRING:" + gisstr,e);
		}	
		
	}
	
	private void emitCountingData(Tuple input) {
		Date date = new Date();
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmm");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		//SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH");
		SimpleDateFormat sdf_min  = new SimpleDateFormat("mm");
		String dateStr_min = sdf_min.format(date);
		
		String dateStr_h = sdf_h.format(date);
		String dateStr_d = sdf_d.format(date);
		
		
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);
		
		SiteEvaluateMain evaluate = new SiteEvaluateMain(zooserver);
		boolean IsTime = false;
		if(System.currentTimeMillis() - _updateTime >= 1800*1000L) {
			IsTime = true;
			_updateTime = System.currentTimeMillis();
			//1小时获取一次新的入库IP配置
//			Slave s = new Slave(zooserver);
//			_ImportIPSet = s.GetIPFilterForLog();
		}
		
		for(String key : DestIPOCTETS.keySet()) {
			
			String IDC = key.split("\\|",-1)[0];
			String destIP = key.split("\\|",-1)[1];
			
			long avgPvalue = DestIPOCTETS.get(key);
			long nPacket = DestIPPACKET.get(key);
			String cf = "cf";
			

			int nMin5 = (Integer.parseInt(dateStr_min) / 5) * 5;
			String sMin5 = String.format("%02d", nMin5);   
			String dateStr5 = dateStr_h + sMin5;
			String rowkey5 = String.format("%s_%s", destIP, dateStr5);
			
			String tbName5 = ApolloProperties.SDS_SESSION_STAT_5MIN;
			hbaseInstance.Add(tbName5, rowkey5, cf,  "DESTIP", destIP);
			hbaseInstance.Add(tbName5, rowkey5, cf,  "REPORTTIME", dateStr5);
			hbaseInstance.Add(tbName5, rowkey5, cf,  "PV", avgPvalue);
			hbaseInstance.Add(tbName5, rowkey5, cf,  "PVNUM", avgPvalue);
			
			SimpleDateFormat dfDay = new SimpleDateFormat("yyyyMMdd");
			Date dStartTimeDay = new Date();
			String strDateDay = dfDay.format(dStartTimeDay);
			
			Calendar c = Calendar.getInstance();
			c.add(Calendar.DATE, 1);// 1 minute ago
			String strEndDay = dfDay.format(c.getTime());
			
			if(IsTime) {
				evaluate.EvaluateFunction(destIP, strDateDay, strEndDay, avgPvalue);
					
				String statRowKeyD = String.format("%s_%s", destIP, strDateDay);
			    String abnStatTbNameDay = ApolloProperties.SDS_ABN_SESS_STAT_D;

			    if(_mapServers != null && _mapServers.containsKey(destIP)) {
			    	ServerInfo serinfo = _mapServers.get(destIP);
			    	hbaseInstance.Add(abnStatTbNameDay, statRowKeyD, cf, "ACCESSTYPE", serinfo.getAccessType());
			    	hbaseInstance.Add(abnStatTbNameDay, statRowKeyD, cf, "LOCATION", serinfo.getServerAddress());
			    } else {
			    	hbaseInstance.Add(abnStatTbNameDay, statRowKeyD, cf, "ACCESSTYPE", "NONE");
			    	hbaseInstance.Add(abnStatTbNameDay, statRowKeyD, cf, "LOCATION", "NONE");
			    }
			}
			
	    	collector.emit(new Values(IDC,"PV",avgPvalue, nPacket));
	    	collector.emit("DELAY",new Values(IDC,destIP, "PV", avgPvalue, nPacket, 0l));
		}
		
		DestIPOCTETS.clear();
		IsTime = false;
		
		
		
		for(Entry<String, HashMap<String,Long>> destIPMap : _mapSourceArea.entrySet()) {
			String destIP = destIPMap.getKey();
			HashMap<String,Long> sourceAreas = destIPMap.getValue();
			
			String tbName = ApolloProperties.SDS_SESSION_AREA_D;
			for(Entry<String, Long> sourceArea : sourceAreas.entrySet()) {
				String sourceKey = sourceArea.getKey();
				String sourceID = sourceKey.split("@",-1)[0];
				String sourceName = sourceKey.split("@",-1)[1];
				
				String rowkey = String.format("%s_%s_%s",destIP, dateStr_d, sourceID);
				String cf = "cf";
				
				hbaseInstance.Add(tbName, rowkey, cf,  "REPORTTIME", dateStr_d);
				hbaseInstance.Add(tbName, rowkey, cf,  "DESTIP", destIP);
				hbaseInstance.Add(tbName, rowkey, cf,  "SOURCEAREA", sourceName);
				hbaseInstance.incerment(tbName, rowkey, "cf:PV", sourceArea.getValue());
			}
			
		}
		
		_mapSourceArea.clear();
//		evaluate.CloseConnection();
		
	}





	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(input)) {//满足时间条件
				Logger.getLogger(BaseStatSumBolt.class).info(String.format("###EMIT BaseStatSumBolt START [%s]",input.getSourceStreamId()));
				emitCountingData(input);  //计算，入库
				Logger.getLogger(BaseStatSumBolt.class).info(String.format("###EMIT BaseStatSumBolt END [%s]",input.getSourceStreamId()));
				collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				collector.ack(input);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			Logger.getRootLogger().error(e);
			throw new FailedException("UDIIBaseStatSumBolt Exception",e);
		}	
	}

	
	public boolean isTickTuple(Tuple tuple) {
		return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(
		        Constants.SYSTEM_TICK_STREAM_ID);   
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		String dbJson = stormConf.get("dbjson_").toString(); //oracle
		ApolloConfig ap = new ApolloConfig(dbJson);
		//获取到服务器配置信息	
		_mapServers = ap.ServerInfos();
		_updateTime = System.currentTimeMillis();
	}
}
